package com.itheima.rabbitmq.producer.config;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.time.LocalDateTime;
import java.util.Date;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

/**
 * @program: springbootRabbitmq->ProducerConfig
 * @description:~
 * @author: G_Y
 * @create: 2019-08-23 18:51
 **/
@Component
public class CallBackListener implements RabbitTemplate.ConfirmCallback/*, RabbitTemplate.ReturnCallback */{
    @Autowired
    private RabbitTemplate rabbitTemplate;
    private Logger logger = LoggerFactory.getLogger(CallBackListener.class);
    @Autowired
    private MessageSender messageSender;

    @PostConstruct
    private void init() {
        rabbitTemplate.setConfirmCallback(this);
//        rabbitTemplate.setReturnCallback(this);
//        rabbitTemplate.setReplyTimeout(1000);
//        rabbitTemplate.setReceiveTimeout(1000);
//        rabbitTemplate.setChannelTransacted(true);  // 设置通道开启事务支持  重量级事务不推荐使用
    }

    // 退回(失败) 回调
//    @Override
//    public void returnedMessage(Message message, int replyCode, String replyText, String exchange,
//                                String routingKey) {
//        logger.info("消息被退回 {}", message.getBody().toString());
//        logger.info("exchange=" + exchange);
//        logger.info("routingKey=" + routingKey);
////        System.out.println(message.getMessageProperties().getCorrelationId() + " 发送失败");
//        rabbitTemplate.convertAndSend(exchange, routingKey, message.getBody());
//        // 存储...
//    }

    // 消息进入Exchane、持久化完成，通知 成功/失败
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        String id = correlationData.getId();
        System.out.println("ack---------id = " + id + "=====" + ack);
        if (ack) {
            logger.info("消息已确认 cause:{} - {}", cause, correlationData);

            // 根据id 取信号灯
            Semaphore semphone = messageSender.getSemphone(id);

            // 先查询数据库
            // 有就直接操作 更新

            if(semphone == null) {
                logger.warn("ACK TRUE MQ ERROR correlationDataId is : " + id);
            }

            logger.warn("ACK TRUE  semphone.acquire BEGIN correlationDataId={}，timestamp={}",id, LocalDateTime.now());
            try {
                semphone.tryAcquire(3, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            logger.warn("ACK TRUE  semphone.acquire END correlationDataId={}，timestamp={}",id, LocalDateTime.now());

            // TODO 查询一下，有数据则更新操作，没有数据则这个 地方 不用管  (因为 ACK 为 true，所以不用管)
            // 结合定时任务最大重试次数，即可

            // 更新 一下 发送状态 为发送成功 TODO

            // 操作完毕 则 移除 semphone
            messageSender.removeSemphone(id);
        } else {
            logger.info("消息未确认 cause:{} - {}", cause, correlationData);
            Semaphore semphone = messageSender.getSemphone(id);
            if(semphone == null) {
                logger.error("MQ ERROR correlationDataId is : " + id);// TODO 查询数据库
            }
            try {
                logger.warn("semphone.acquire BEGIN correlationDataId={}，timestamp={}",id, LocalDateTime.now());
                boolean b = semphone.tryAcquire(5, TimeUnit.SECONDS);
                logger.warn("semphone.acquire END correlationDataId={}，timestamp={}",id, LocalDateTime.now());

                if(b) {
                    // 灯亮了
                    // TODO 查询数据库

                    // TODO 重发MQ 更新发送时间
                } else {
                    // 没有等到灯亮
                    logger.error("MQ ERROR semphone.acquire TIMEOUT DELETED correlationDataId is : {} , e ={} ", id);
                }
                // 移除信号灯
                messageSender.removeSemphone(id);
            } catch (InterruptedException e) {
                e.printStackTrace();
                logger.error("MQ ERROR semphone.acquire TIMEOUT DELETED correlationDataId is : {} , e ={} ",id, e.getMessage(), e);
            }
        }
    }
}
